package net.bwie.realtime.dwd.douyin.logs.job;

import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 交易直播纬度数据清洗
 */

public class LiveRoomInfoCleanJob {

    public static void main(String[] args) throws Exception {

        //1-执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2. 从kafka的ods直播纬度数据消费
        DataStream<String> rawStream = KafkaUtil.consumerKafka(env, "ods_live_room_info");
        rawStream.print("row");

        //3-数据清洗-json解析+字段校验+过滤
        SingleOutputStreamOperator<String> cleanStream = rawStream.map(new MapFunction<String, String>() {
                    @Override
                    public String map(String s) throws Exception {
                        //json字符串解析
                        try{
                            return JSONObject.parseObject(s).toJSONString();
                        } catch (Exception e) {
                            return null;
                        }
                    }
                })
                //过滤空值
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        if(s != null){
                            return true;
                        }
                        return false;
                    }
                })
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(s);
                        if(
                            //containsKey:判断key是否存在，存在再判断
                                jsonObject.containsKey("live_session_id") &&
                                        jsonObject.containsKey("anchor_id")
                        ){
                            return true;
                        }
                        return false;
                    }
                })
                // 数值字段校验
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(s);
                        if(
                            //停留时长不为null,>=0
                                jsonObject.getDouble("live_duration_sec") != null &&
                                        jsonObject.getDouble("live_duration_sec") >= 0
                        ){
                            return true;
                        }
                        return false;
                    }
                })
                // 枚举校验
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                            JSONObject jsonObject = JSONObject.parseObject(s);
                            String start = jsonObject.getString("start_time");
                            String end = jsonObject.getString("end_time");
                            return isValidTimeRange(start, end);
                    }

                    private boolean isValidTimeRange(String start, String end) {
                        //如果end_time在start_time之后数据保留,反之过滤掉
                        //如果时间字段为空，或者格式解析失败，就会抛异常
                        try {
                            if (start == null || end == null) {
                                return false;
                            }
                            java.sql.Timestamp startTs = java.sql.Timestamp.valueOf(start);
                            java.sql.Timestamp endTs = java.sql.Timestamp.valueOf(end);
                            return endTs.after(startTs); // 等价于 end_time > start_time
                        } catch (Exception e) {
                            // 如果格式错误，默认放行
                            return true;
                        }
                    }
                });

        //4. 打印清洗后的数据（后续可以写入dwdkafka）
        cleanStream.print("cleanLiveInfoDetail");

        KafkaUtil.producerKafka(cleanStream,"dwd_room_info");
        //5. 执行任务
        env.execute("LiveRoomInfoCleanJob");
    }

}

